package belf.migrate.plugin.postgresql;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.exception.*;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.ExceptionUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

@Slf4j
public class PostgreSqlCatalog extends AbstractJdbcCatalog {

    private static final String QUOTE = "\"";

    public PostgreSqlCatalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.POSTGRESQL, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getDatabaseAndTableName();
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    /**
     * 获取某个数据库或者模式下所有的表
     * @param database
     * @param schema
     * @return
     */
    @Override
    public String getListTableSQL(String database, String schema) {
        return """
                SELECT t.table_schema,\s
                       t.table_name as TABLE_NAME
                FROM information_schema.tables t
                LEFT JOIN pg_inherits i\s
                ON t.table_name = inhrelid::regclass::text
                WHERE t.table_schema = '%s'
                  AND t.table_type = 'BASE TABLE'
                  AND i.inhrelid IS NULL
                ORDER BY t.table_name
               """.formatted(schema);
    }

    /**
     * 获取所有的数据库或者schema
     * @return
     */
    @Override
    public String getListDatabaseSQL() {
        return """
                SELECT nspname AS schema_name
                FROM pg_namespace
                WHERE nspname NOT IN ('pg_catalog', 'information_schema')
                AND nspname NOT LIKE 'pg_toast%'
                ORDER BY nspname
           """;
    }

    @Override
    public String getListViewsSQL(String schemaName) {
        return String.format("""
                SELECT table_name AS view_name
                FROM information_schema.views
                WHERE table_schema = '%s'
                """, schemaName);
    }

    @Override
    public String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return getListViewsSQL(schemaName);
    }

    @Override
    public String getListFunctionSQL(String schemaName) {
        return String.format("""
                SELECT p.proname AS FUNCTION_NAME
                FROM pg_proc p
                JOIN pg_namespace n ON p.pronamespace = n.oid
                WHERE n.nspname = '%s'
                and prokind in ('f', 'a')
                """, schemaName);
    }

    @Override
    public String getListTriggerSQL(String schemaName) {
        // TODO postgresql 是表级触发器，schema下存在同名问题
        return null;
    }

    @Override
    public String getListIndexSQL(String schemaName) {
        String sqlTemplate = """
            SELECT i.relname AS index_name
            FROM pg_index idx
            JOIN pg_class i ON i.oid = idx.indexrelid
            JOIN pg_class t ON t.oid = idx.indrelid
            JOIN pg_namespace n ON t.relnamespace = n.oid
            WHERE n.nspname = '%s'
            """;
        return String.format(sqlTemplate, schemaName);
    }

    @Override
    public String getListTypeSQL(String defaultDatabase) {
        return "SELECT t.typname as TYPE_NAME\n" +
               "FROM pg_catalog.pg_type t\n" +
               "LEFT OUTER JOIN pg_catalog.pg_type et ON et.oid=t.typelem \n" +
               "LEFT OUTER JOIN pg_catalog.pg_class c ON c.oid=t.typrelid\n" +
               "LEFT OUTER JOIN pg_catalog.pg_description d ON t.oid=d.objoid\n" +
               "WHERE t.typname IS NOT NULL\n" +
               "AND (c.relkind IS NULL OR c.relkind = 'c') AND (et.typcategory IS NULL OR et.typcategory <> 'C')\n" +
               "and t.typrelid <> 0";
    }

    @Override
    public String getListProcedureSQL(String defaultDatabase) {
        return String.format("""
           SELECT p.proname AS PROCEDURE_NAME
           FROM pg_proc p
           JOIN pg_namespace n ON p.pronamespace = n.oid
           WHERE n.nspname = '%s'
           and prokind = 'p'
           """, defaultDatabase);
    }

    @Override
    public String getExistDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getTableName();
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getSchemaAndTableName();
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName)
            throws CatalogException {
        String sql = String.format("""
                SELECT
                relname AS TABLE_NAME,
                n_live_tup AS TABLE_ROWS,
                ROUND(pg_total_relation_size(relid) / 1024.0 / 1024.0, 2) AS TABLE_SIZE
                FROM
                pg_stat_user_tables
                WHERE
                schemaname = '%s'
                """, schemaName);
        try (PreparedStatement ps =
                     getConnection(defaultUrl)
                             .prepareStatement(sql);
             ResultSet rs = ps.executeQuery()) {

            List<TableInfoModel> tables = new ArrayList<>();
            while (rs.next()) {
                TableInfoModel table = new TableInfoModel();
                table.setTableName(rs.getString("TABLE_NAME"));
                table.setRows(rs.getLong("TABLE_ROWS"));
                table.setSize(rs.getLong("TABLE_SIZE"));
                tables.add(table);
            }

            return tables;
        } catch (Exception e) {
            throw new CatalogException(
                    String.format("Failed listing table in catalog %s", catalogName), e);
        }
    }

    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) throws CatalogException {
        TableInfoModel table = new TableInfoModel();
        try {
            String sql = String.format("""
                SELECT
                relname AS TABLE_NAME,
                n_live_tup AS TABLE_ROWS,
                ROUND(pg_total_relation_size(relid) / 1024.0 / 1024.0, 2) AS TABLE_SIZE
                FROM
                pg_stat_user_tables
                WHERE
                schemaname = '%s'
                and relname = '%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
            log.info("Query mysql table info: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setSize(resultSet.getLong("TABLE_SIZE"));
                table.setComment(resultSet.getString("TABLE_COMMENT"));
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("获取表的信息失败！", e);
        }
        return table;
    }

    @Override
    public String getListSequenceSQL(String schema) {
        return String.format("""
                SELECT
                sequence_schema,
                SEQUENCE_NAME
                FROM
                information_schema.sequences
                WHERE
                sequence_schema = '%s'
                """, schema);
    }

    @Override
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
        String sql = String.format("""
                SELECT
                cols.column_name AS COLUMN_NAME,
                cols.column_default AS COLUMN_DEFAULT,
                cols.numeric_scale AS DATETIME_PRECISION,
                pgd.description AS COLUMN_COMMENT
                FROM
                information_schema.columns cols
                LEFT JOIN
                pg_catalog.pg_statio_all_tables stat ON stat.schemaname = cols.table_schema AND stat.relname = cols.table_name
                LEFT JOIN
                pg_catalog.pg_description pgd ON pgd.objoid = stat.relid AND pgd.objsubid = cols.ordinal_position
                WHERE
                cols.table_schema = '%s'
                AND cols.table_name = '%s'
                ORDER BY
                cols.ordinal_position
                """, tablePath.getSchemaName(), tablePath.getTableName());

        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                String columnDefault = rs.getString("COLUMN_DEFAULT");
                int datetimePrecision = rs.getInt("DATETIME_PRECISION");
                String columnComment = rs.getString("COLUMN_COMMENT");
                Optional<ColumnMetaData> possibleColumn = columns.stream().filter(new Predicate<ColumnMetaData>() {
                    @Override
                    public boolean test(ColumnMetaData columnMetaData) {
                        return columnMetaData.getColumnName().equalsIgnoreCase(columnName);
                    }
                }).findFirst();
                if (possibleColumn.isPresent()) {
                    ColumnMetaData columnMetaData = possibleColumn.get();
                    ColumnExtra columnExtra = new ColumnExtra(columnDefault, null, columnComment, datetimePrecision);
                    columnMetaData.setExtra(columnExtra);
                }
            }
        } catch (SQLException e) {
            log.error("Error on query db2 column extra information:\n{}", ExceptionUtils.toString(e));
        }
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl = defaultUrl;

        // 查询 PostgreSQL 数据库中指定表的索引，排除主键索引
        String sql = String.format("""
            SELECT
                c.relname AS index_name,
                i.indisunique AS is_unique,
                a.attname AS column_name,
                (array_position(i.indkey, a.attnum)) AS seq_in_index,
                am.amname AS index_type,
                CASE i.indoption[array_position(i.indkey, a.attnum) - 1] & 1
                    WHEN 0 THEN 'ASC'
                    ELSE 'DESC'
                END AS index_order
            FROM pg_index i
            JOIN pg_class c ON c.oid = i.indexrelid
            JOIN pg_class t ON t.oid = i.indrelid
            JOIN pg_namespace n ON n.oid = t.relnamespace
            JOIN pg_attribute a ON a.attnum = ANY (i.indkey) AND a.attrelid = t.oid
            JOIN pg_am am ON c.relam = am.oid
            WHERE
                n.nspname = '%s'
                AND t.relname = '%s'
                AND NOT i.indisprimary
            ORDER BY index_name, seq_in_index
            """, tablePath.getSchemaName(), tablePath.getTableName());

        try {
            Connection conn = getConnection(dbUrl);
            Statement statement = conn.createStatement();

            DatabaseMetaData metaData = conn.getMetaData();
            List<IndexKey> defaultIndexKeys = CatalogUtils.getIndexKeys(metaData, tablePath);

            // index name -> index
            Map<String, IndexKeyExt> indexKeyMap = new HashMap<>();

            try (ResultSet rs = statement.executeQuery(sql)) {
                log.debug("query index:{}", sql);
                while (rs.next()) {
                    boolean unique = rs.getBoolean("is_unique");
                    String indexName = rs.getString("index_name");
                    int sequenceInIndex = rs.getInt("seq_in_index");
                    String columnName = rs.getString("column_name");
                    String indexOrder = rs.getString("index_order");
                    String indexTypeName = rs.getString("index_type");

                    List<IndexKeyExt.ColumnWithSequence> columns = new ArrayList<>();
                    columns.add(new IndexKeyExt.ColumnWithSequence(columnName, sequenceInIndex));
                    IndexKeyExt indexKeyExt = new IndexKeyExt(
                            columns,
                            indexName,
                            unique,
                            indexOrder,
                            indexTypeName);
                    IndexKeyExt existOne = indexKeyMap.get(indexName);
                    if (existOne != null) {
                        existOne.getColumns().addAll(indexKeyExt.getColumns());
                    } else {
                        indexKeyMap.put(indexName, indexKeyExt);
                    }

                    log.debug("got index: {}", indexKeyExt);
                }
            }

            Iterator<IndexKey> it = defaultIndexKeys.iterator();
            while (it.hasNext()) {
                IndexKey indexKey = it.next();
                IndexKeyExt indexKeyExt = indexKeyMap.get(indexKey.getIndexName());
                if (null != indexKeyExt) {
                    indexKey.setColumnNames(indexKeyExt.getSortedColumnNames());
                    indexKey.setIndexTypeName(indexKeyExt.getIndexTypeName());
                    if (indexKeyExt.getIndexOrder() != null
                        && "ASC".equalsIgnoreCase(indexKeyExt.getIndexOrder())) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.ASC);
                    } else if (indexKeyExt.getIndexOrder() != null
                               && "DESC".equalsIgnoreCase(indexKeyExt.getIndexOrder())) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.DESC);
                    }
                } else {
                    it.remove();
                }
            }

            statement.close();
            return defaultIndexKeys;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        // TODO: 暂未找到获取表的ddl的方法
        return null;
    }

    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {

        String sql = "SELECT \n" +
                       "    format(\n" +
                       "        'CREATE OR REPLACE VIEW %I.%I AS %s',\n" +
                       "        schemaname,\n" +
                       "        viewname,\n" +
                       "        definition\n" +
                       "    ) AS VIEW_DDL\n" +
                       "FROM pg_views ";

        sql = sql + String.format("""
                WHERE schemaname = '%s'
                AND viewname = '%s'
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("VIEW_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new ViewNotExistException(DatabaseIdentifier.DB_2, tablePath);
        }
        return null;
    }


    @Override
    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        String query = String.format("""
                        SELECT pg_get_functiondef(p.oid) AS PROCEDURE_DDL
                                            FROM pg_proc p
                                            JOIN pg_namespace n ON p.pronamespace = n.oid
                                            WHERE n.nspname = 'public'\s
                                              AND p.proname = 'insert_into_table';
                                            
                """, tablePath.getSchemaName(), tablePath.getTableName());
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("PROCEDURE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create procedure : \n{}", sql);
            throw new ProcedureNotExistException(DatabaseIdentifier.DB_2, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException {
        // TODO: 暂未找到获取trigger的ddl的方法
        return null;
    }

    @Override
    public String showCreateType(TablePath tablePath) throws TypeNotExistException {

        String sql = String.format("""
                WITH type_columns AS (
                    SELECT
                        a.attname AS column_name,
                        format_type(a.atttypid, a.atttypmod) AS data_type,
                        a.attnotnull AS not_null
                    FROM
                        pg_attribute a
                        JOIN pg_type t ON a.attrelid = t.typrelid
                        JOIN pg_namespace n ON t.typnamespace = n.oid
                    WHERE
                        t.typname = '%s'  -- 替换为你的类型名
                        AND n.nspname = '%s'    -- 替换为你的 schema 名
                        AND a.attnum > 0            -- 排除系统列
                        AND NOT a.attisdropped      -- 排除已删除的列
                )
                SELECT
                    'CREATE TYPE public.film_summary AS (' || string_agg(
                        column_name || ' ' || data_type || CASE WHEN not_null THEN ' NOT NULL' ELSE '' END,
                        ', '
                    ) || ');'
                AS TYPE_DDL
                FROM type_columns
                """, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("TYPE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create type: \n{}", sql);
            throw new TypeNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateSeq(TablePath tablePath) throws SequenceNotExistException {
        String sql = String.format("""
                SELECT
                    'CREATE SEQUENCE ' || quote_ident(n.nspname) || '.' || quote_ident(c.relname) || E'\\n' ||
                    '    START WITH ' || s.seqstart || E'\\n' ||
                    '    INCREMENT BY ' || s.seqincrement || E'\\n' ||
                    '    MINVALUE ' || s.seqmin || E'\\n' ||
                    '    MAXVALUE ' || s.seqmax || E'\\n' ||
                    '    CACHE ' || s.seqcache || E'\\n' ||
                    CASE WHEN s.seqcycle THEN '    CYCLE' ELSE '    NO CYCLE' END || ';'
                AS SEQ_DDL
                FROM
                    pg_class c
                    JOIN pg_namespace n ON c.relnamespace = n.oid
                    JOIN pg_sequence s ON c.oid = s.seqrelid
                WHERE
                    n.nspname = '%s' -- 替换为序列所属的 schema 名称
                    AND c.relname = '%s' -- 替换为序列名称
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("SEQ_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create sequence: \n{}", sql);
            throw new SequenceNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        String ddl = "";
        String sql = String.format("""
                SELECT pg_get_functiondef(p.oid) AS FUNC_DDL
                FROM pg_proc p
                JOIN pg_namespace n ON p.pronamespace = n.oid
                WHERE n.nspname = '%s'  -- 替换为目标模式名称
                AND p.proname = '%s';  -- 替换为目标函数名称
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            boolean hasResults = statement.execute(sql);
            while (hasResults) {
                try (ResultSet resultSet = statement.getResultSet()) {
                    while (resultSet.next()) {
                        if (resultSet.getString("FUNC_DDL") != null) {
                            ddl = resultSet.getString("FUNC_DDL");
                        }
                    }
                }
                // 检查是否有下一个结果集
                hasResults = statement.getMoreResults();
            }
        } catch (SQLException e) {
            log.error("Error while show create function: \n{}", sql);
            throw new FunctionNotExistException(DatabaseIdentifier.DB_2, tablePath, e);
        }
        return ddl;
    }

    @Override
    public String showCreateIndex(TablePath tablePath) throws IndexNotExistException {
        String sql = String.format("""
                SELECT pg_get_indexdef(i.oid) AS INDEX_DDL
                FROM pg_index x
                JOIN pg_class i ON x.indexrelid = i.oid
                JOIN pg_namespace n ON i.relnamespace = n.oid
                WHERE n.nspname = '%s'  -- 替换为目标模式名称
                  AND i.relname = '%s';  -- 替换为目标索引名称
                """, tablePath.getSchemaName(), tablePath.getTableName());
        try {
            return getDDL(sql, "INDEX_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create index: \n{}", sql);
            throw new IndexNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
    }

    @Override
    public String getDDL(String sql, String ddlName) throws CatalogException {
        String ddl = "";
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);
        try (Statement statement = connection.createStatement()) {
            // Will there exist concurrent drop for one table?
            boolean hasResults = statement.execute(sql);
            while (hasResults) {
                try (ResultSet resultSet = statement.getResultSet()) {
                    while (resultSet.next()) {
                        if (resultSet.getString(ddlName) != null) {
                            ddl = resultSet.getString(ddlName);
                        }
                    }
                }
                // 检查是否有下一个结果集
                hasResults = statement.getMoreResults();
            }
        } catch (SQLException e) {
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, String.format("Failed executeSql error: \n%s", sql), e);
        }
        return ddl;
    }

    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
                SELECT
                c.conname AS CONSTRAINT_NAME,
                pg_get_constraintdef(c.oid) AS constraint_definition,
                a.attname AS COLUMN_NAME
                FROM
                pg_constraint c
                JOIN
                pg_class t ON c.conrelid = t.oid
                JOIN
                pg_namespace n ON t.relnamespace = n.oid
                JOIN
                unnest(c.conkey) AS colnum(colnum) -- 解包 conkey 数组
                ON true
                JOIN
                pg_attribute a ON a.attnum = colnum.colnum AND a.attrelid = t.oid
                WHERE
                n.nspname = '%s'  -- 替换为目标模式名称
                AND t.relname = '%s'   -- 替换为目标表名称
                AND c.contype = 'u'                 -- 仅查询唯一键约束
                ORDER BY
                c.conname, colnum.colnum;
                """, tablePath.getSchemaName(), tablePath.getTableName());
        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String column = rs.getString("COLUMN_NAME");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "DB2查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
                SELECT
                c.conname AS constraint_name,
                t.relname AS table_name,
                n.nspname AS table_schema,
                'CHECK' AS constraint_type,
                pg_get_constraintdef(c.oid) AS check_condition
                FROM
                pg_constraint c
                JOIN
                pg_class t ON c.conrelid = t.oid
                JOIN
                pg_namespace n ON t.relnamespace = n.oid
                WHERE
                n.nspname = '%s'  -- 替换为目标模式名称
                AND t.relname = '%s'  -- 替换为目标表名称
                AND c.contype = 'c'  -- 'c' 表示 CHECK 约束
                """, tablePath.getSchemaName(), tablePath.getTableName());
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String condition = rs.getString("CHECK_CONDITION");
                    checkConstraints.add(new CheckConstraint(tablePath.getTableName(),
                            constraintName,
                            condition));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Check约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }


    @Override
    public List<UserDefFunction> listFunctions(String schema, List<String> functionNames) throws CatalogException {
        String sql = String.format("""
                SELECT pg_get_functiondef(p.oid) AS FUNC_DDL,
                p.proname as FUNCNAME
                FROM pg_proc p
                JOIN pg_namespace n ON p.pronamespace = n.oid
                WHERE n.nspname = '%s'  -- 替换为目标模式名称
                AND p.prokind = 'f'  -- 过滤掉聚合函数
                """, schema);
        if (null != functionNames && !functionNames.isEmpty()) {
            List<String> quotedFuncNames = new ArrayList<>();
            for (String funcName: functionNames) {
                quotedFuncNames.add("'" + funcName + "'");
            }
            sql += String.format(" AND p.proname IN (%s)", String.join(",", quotedFuncNames));
        }
        List<UserDefFunction> userDefFunctions = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefFunction userDefFunction = new UserDefFunction();
                    String functionName = rs.getString("FUNCNAME");
                    userDefFunction.setFunctionName(functionName);
                    userDefFunction.setDdl(rs.getString("FUNC_DDL"));
                    userDefFunctions.add(userDefFunction);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_LIST_ERROR, "查询所有自定义函数的DDL失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefFunctions;
    }


    @Override
    protected String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + tablePath.getSchemaAndTableName();
    }

    @Override
    public String getUrlFromDatabaseName(String database) {
        return defaultUrl;
    }

}
